package com.leinao.resourcesInfo.service.impl;

import java.math.BigDecimal;
import java.net.URI;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.leinao.resourcesInfo.dao.PaiResourcesOrderMapper;
import com.leinao.resourcesInfo.dao.PaiUserResourceInfoMapper;
import com.leinao.resourcesInfo.model.PaiUserResourceInfo;
import com.leinao.resourcesInfo.model.UserJobBandwidth;
import com.leinao.resourcesInfo.service.UserJobBandWidthService;
import com.leinao.resourcesInfo.service.UserResourcesInfoService;
import com.leinao.user.dao.UserBasicInfoMapper;

/**
*
* 定时统计多资源情况入库
* created by wangshoufa
* 2018年8月28日 下午5:12:57
*
*/

@Service
public class UserResourcesInfoServiceImpl implements UserResourcesInfoService {
	
	private static final Logger logger = LoggerFactory.getLogger(UserResourcesInfoServiceImpl.class);
	
	@Autowired
	private RestTemplate restTemplate;
	
	@Value("${remote.loadJobUrl}")
	private String loadJobUrl;
	
	@Value("${remote.loginUrl}")
	private String loginUrl;
	
	@Value("${remote.frameworksUrl}")
	private String frameworksUrl;
	
	@Value("${remote.prometheus.url}")
	private String prometheusUrl;
	//?query=container_NetIn{container_label_PAI_USER_NAME=\"_e2eb9236063744fe84bba63f89f56fad\"}
	
	@Autowired
	private PaiUserResourceInfoMapper paiUserResourceInfoMapper;
	
	@Autowired
	private PaiResourcesOrderMapper paiResourcesOrderMapper;
	
	@Autowired
	private UserBasicInfoMapper userBasicInfoMapper;
	
	@Autowired
	private UserJobBandWidthService userJobBandWidthService;
	
	private static List<UserJobBandwidth> jobBandWidthList;
	
	@SuppressWarnings("unused")
	@Override
	public int userResourcesInfoTask() {
		String token = authToken();
		if(StringUtils.isNotBlank(token)) {
			List<String> userIds = userBasicInfoMapper.queryBasicUserInfo();
			try {
				HttpHeaders headers = new HttpHeaders();
    	        MediaType type = MediaType.parseMediaType("application/json; charset=UTF-8");
    	        headers.setContentType(type);
		        headers.add("Accept", MediaType.APPLICATION_JSON.toString());
		        headers.add("Authorization", "Bearer " + token);
    	        HttpEntity<String> formEntity = new HttpEntity<String>(null, headers);
				String body = "";
				try {
					ResponseEntity<String> responseEntity = restTemplate.exchange(loadJobUrl, HttpMethod.GET, formEntity, String.class);
					logger.info("request getAllJobInfo return: " + responseEntity);
					if(responseEntity.getStatusCodeValue() == 200) {
						logger.info("get jobs success");
						body = responseEntity.getBody();
					}
				} catch (Exception e) {
					logger.error("get jobs error");
					logger.error(e.getMessage(), e);
					return 0;
				}
				
				if(userIds!=null && userIds.size()>0) {
					List<PaiUserResourceInfo> insertList = Lists.newArrayList();
					List<PaiUserResourceInfo> updateList = Lists.newArrayList();
					
					for(String userId : userIds) {
						BigDecimal allGpuResource = new BigDecimal(0);
						BigDecimal allCpuResource = new BigDecimal(0);
						BigDecimal allMemoryResource = new BigDecimal(0);
						BigDecimal allDiskResource = new BigDecimal(0);
						BigDecimal allTapeWidthResource = new BigDecimal(0);
						BigDecimal memoryResource = new BigDecimal(0);
						BigDecimal cpuResource = new BigDecimal(0);
						BigDecimal gpuResource = new BigDecimal(0);
						BigDecimal diskResource = new BigDecimal(0);
						
						//当前用户的各项总资源
						Map gpuMap = paiResourcesOrderMapper.selectUserResByResId(userId, "0");
						if(gpuMap != null && gpuMap.containsKey("user_id")) {
							allGpuResource = (BigDecimal) gpuMap.get("allTime");
						}
						else {
							allGpuResource = new BigDecimal(0);
						}
						logger.info(allGpuResource.toString());
						Map cpuMap = paiResourcesOrderMapper.selectUserResByResId(userId, "1");
						if(cpuMap != null && cpuMap.containsKey("user_id")) {
							allCpuResource = (BigDecimal) cpuMap.get("allTime");
						}
						else {
							allCpuResource = new BigDecimal(0);
						}
						Map memoryMap = paiResourcesOrderMapper.selectUserResByResId(userId, "2");
						if(memoryMap != null && memoryMap.containsKey("user_id")) {
							allMemoryResource = (BigDecimal) memoryMap.get("allTime");
						}
						else {
							allMemoryResource = new BigDecimal(0);
						}
						Map diskMap = paiResourcesOrderMapper.selectUserResByResId(userId, "3");
						if(diskMap != null && diskMap.containsKey("user_id")) {
							allDiskResource = (BigDecimal) diskMap.get("allTime");
						}
						else {
							allDiskResource = new BigDecimal(0);
						}
						Map tapeWidthMap = paiResourcesOrderMapper.selectUserResByResId(userId, "4");
						if(tapeWidthMap != null && tapeWidthMap.containsKey("user_id")) {
							allTapeWidthResource = (BigDecimal) tapeWidthMap.get("allTime");
						}
						else {
							allTapeWidthResource = new BigDecimal(0);
						}
						JSONArray jsonArray = JSON.parseArray(body);
						for(int i = 0; i < jsonArray.size(); i++) {
							JSONObject jsonObject = (JSONObject) jsonArray.get(i);
							String jobName = jsonObject.getString("name");
							String user_id = jsonObject.getString("userId");
							Long createdTime = (Long) jsonObject.get("createdTime");
							Object objTime = jsonObject.get("completedTime");
							Long completedTime;
							if(objTime != null) {
								completedTime = (Long) objTime;
							}
							else {
								completedTime = System.currentTimeMillis();
							}
							//匹配用户
							if(StringUtils.isNotBlank(userId) && StringUtils.isNotBlank(user_id) && userId.equals(user_id)) {
								ResponseEntity<String> result = restTemplate.getForEntity(frameworksUrl + jobName, String.class);
								if(result.getStatusCode().OK.value() == 200) {
									//logger.info("请求获取用户任务明细返回包:" + JSON.toJSON(result));
									JSONObject reponse = JSON.parseObject(result.getBody());
									Object object = reponse.get("aggregatedFrameworkStatus");
									Object obj = JSON.parseObject(object.toString()).get("frameworkStatus");
									Long frameworkCreatedTimestamp = (Long) JSON.parseObject(obj.toString()).get("frameworkCreatedTimestamp");
									Long frameworkCompletedTimestamp;
									Object time = JSON.parseObject(obj.toString()).get("frameworkCompletedTimestamp");
									if(time != null) {
										frameworkCompletedTimestamp = (Long) time;
									}
									else {
										frameworkCompletedTimestamp = System.currentTimeMillis();
									}
									double usedTime = Math.round(Math.max(0, frameworkCompletedTimestamp - frameworkCreatedTimestamp) / 1000);
									usedTime = BigDecimal.valueOf(usedTime).divide(new BigDecimal(3600), 4, BigDecimal.ROUND_HALF_UP).doubleValue();
									logger.info("usedTime: " + usedTime);
									
									Object frameworkRequest = JSON.parseObject(reponse.get("aggregatedFrameworkRequest").toString()).get("frameworkRequest");
									Object frameworkDescriptor = JSON.parseObject(frameworkRequest.toString()).get("frameworkDescriptor");
									Object taskRoles = JSON.parseObject(frameworkDescriptor.toString()).get("taskRoles");
									JSONObject array = JSON.parseObject(taskRoles.toString());
									if(array!=null && array.size() > 0) {
										//动态解析taskRole的名称以及数据信息
										Set<String> keys = array.keySet();
										Iterator iterator = keys.iterator();
										while(iterator.hasNext()){
											String key = (String) iterator.next();
											logger.info("taskRoleName key: "+key);
											Object taskRole = array.get(key);
											Integer taskNumber = (Integer) JSON.parseObject(taskRole.toString()).get("taskNumber");
											Object taskService = JSON.parseObject(taskRole.toString()).get("taskService");
											Object resource = JSON.parseObject(taskService.toString()).get("resource");
											Integer gpuNumber = (Integer) JSON.parseObject(resource.toString()).get("gpuNumber");
											Integer cpuNumber = (Integer) JSON.parseObject(resource.toString()).get("cpuNumber");
											Integer memoryMB = (Integer) JSON.parseObject(resource.toString()).get("memoryMB");
											Integer diskMB = (Integer) JSON.parseObject(resource.toString()).get("diskMB");
											logger.info("gpuNumber: " + gpuNumber);
											gpuNumber = taskNumber * gpuNumber;
											cpuNumber = taskNumber * cpuNumber;
											double memory = (double) taskNumber * memoryMB/1024;
											double disk = (double) taskNumber * diskMB/1024;
											double gpuRes = usedTime * gpuNumber;
											gpuResource = BigDecimal.valueOf(gpuRes).add(gpuResource);
											double cpuRes = usedTime * cpuNumber;
											cpuResource = BigDecimal.valueOf(cpuRes).add(cpuResource);
											double memoryRes = usedTime * memory;
											memoryResource = BigDecimal.valueOf(memoryRes).add(memoryResource);
											double diskRes = usedTime * disk;
											diskResource = BigDecimal.valueOf(diskRes).add(diskResource);
										}
									}
								}
							}
						}
						//计算完每一个用户的所有任务的资源统计
						if(cpuResource.compareTo(new BigDecimal(0)) >0) {
							PaiUserResourceInfo resInfo = new PaiUserResourceInfo();
							resInfo.setUserId(userId);
							resInfo.setGpuResource(allGpuResource);
							resInfo.setGpuUsedResource(gpuResource);
							resInfo.setCpuResource(allCpuResource);
							resInfo.setCpuUsedResource(cpuResource);
							resInfo.setMemoryResource(allMemoryResource);
							resInfo.setMemoryUsedResource(memoryResource);
							resInfo.setDiskResource(allDiskResource);
							resInfo.setDiskUsedResource(diskResource);
							resInfo.setBandwidthResource(allTapeWidthResource);
							//是否已经入库
							PaiUserResourceInfo info = paiUserResourceInfoMapper.selectUserResourceInfoByUserId(userId);
							//入库则更新
							if(info!=null && StringUtils.isNumeric(String.valueOf(info.getId()))) {
								logger.info("resInfo: " +resInfo);
								updateList.add(resInfo);
							}
							//初始化
							else {
								insertList.add(resInfo);
							}
						}
					}
					insertList(insertList);
					updateList(updateList);
				}
			}
			catch (RestClientException e) {
				logger.error(e.getMessage(), e);
				return 0;
			}
		}
		return 1;
	}
	
	@Override
	public int userBandWidthResourceTask() {
		List<String> userIds = paiUserResourceInfoMapper.selectUserResourceInfo();
		if(!CollectionUtils.isEmpty(userIds)) {
			List<UserJobBandwidth> insertList = Lists.newArrayList();
			for(String userId : userIds) {
				BigDecimal tapeWidthResource = new BigDecimal(0);
				String body = "";
				try {
					if(!CollectionUtils.isEmpty(jobBandWidthList)) {
						jobBandWidthList.clear();
					}
					//int deleteResult = userJobBandWidthService.deleteRedundant();
					List<UserJobBandwidth> widThList = userJobBandWidthService.queryAll();
					if(!CollectionUtils.isEmpty(widThList)) {
						jobBandWidthList = widThList;
					}
					String url = "{container_label_PAI_USER_NAME=\""+ userId +"\"}";//&start=" + (System.currentTimeMillis()/1000 - 5) + "&end=" + System.currentTimeMillis()/1000 +"&step=30";
					String enCode = URLEncoder.encode(url,"UTF-8");
					URI uri = URI.create(prometheusUrl + enCode);
					ResponseEntity<String> responseEntity = restTemplate.getForEntity(uri, String.class);
					logger.info("request getBandWidthInfo return: " + responseEntity);
					if(responseEntity.getStatusCodeValue() == 200) {
						body = responseEntity.getBody();
						JSONObject object = JSON.parseObject(body);
						if(StringUtils.isNotBlank(object.getString("status")) && "success".equals(object.getString("status"))) {
							Object obj = object.get("data");
							Object resu = JSON.parseObject(obj.toString()).get("result");
							JSONArray result = JSON.parseArray(resu.toString());
							if(result!=null && result.size() >0) {
								for(int i = 0; i< result.size();i++) {
									UserJobBandwidth bandWidth = new UserJobBandwidth();
									Object width = result.get(i);
									Object metric = JSON.parseObject(width.toString()).get("metric");
									String jobName = (String) JSON.parseObject(metric.toString()).get("container_label_PAI_JOB_NAME");
									String taskRoleName = (String) JSON.parseObject(metric.toString()).get("container_label_PAI_CURRENT_TASK_ROLE_NAME");
									String taskIndex = (String) JSON.parseObject(metric.toString()).get("container_env_PAI_TASK_INDEX");
									Object value = JSON.parseObject(width.toString()).get("value");
									JSONArray array = JSON.parseArray(value.toString());
									String bindWidthResoure = (String) array.get(1);
									tapeWidthResource = new BigDecimal(bindWidthResoure);
									bandWidth.setUserId(userId);
									bandWidth.setJobName(jobName);
									bandWidth.setTaskRoleName(taskRoleName);
									bandWidth.setTaskIndex(taskIndex);
									//单位:kb
									bandWidth.setBandWidth(tapeWidthResource.divide((new BigDecimal(1024)), 4, BigDecimal.ROUND_HALF_UP));
									insertList.add(bandWidth);
								}
							}
							
							logger.info("insertList:" + insertList);
							if(!CollectionUtils.isEmpty(insertList)) {
								if(!CollectionUtils.isEmpty(jobBandWidthList)) {
									logger.info("jobBandWidthList:" + jobBandWidthList);
									List<UserJobBandwidth> updateList = Lists.newArrayList();
									List<UserJobBandwidth> newList = Lists.newArrayList();
									
									//比较两次的内存数据，看看有哪些是需要更新的，哪些是需要入库的
									for (int i = 0; i < insertList.size(); i++) {
										for (int j = 0; j < jobBandWidthList.size(); j++) {
											UserJobBandwidth newBand = insertList.get(i);
											UserJobBandwidth band = jobBandWidthList.get(j);
											if(band.getUserId().equals(newBand.getUserId()) &&
													band.getJobName().equals(newBand.getJobName()) &&
													band.getTaskRoleName().equals(newBand.getTaskRoleName()) &&
													band.getTaskIndex().equals(newBand.getTaskIndex()) &&
													band.getBandWidth().compareTo(newBand.getBandWidth()) < 0) {
													updateList.add(newBand);
													break;
											}
											if(band.getUserId().equals(newBand.getUserId()) &&
													band.getJobName().equals(newBand.getJobName()) &&
													band.getTaskRoleName().equals(newBand.getTaskRoleName()) &&
													band.getTaskIndex().equals(newBand.getTaskIndex()) &&
													band.getBandWidth().compareTo(newBand.getBandWidth()) >= 0) {
													logger.info("old >= new  " + band + "   " + newBand);
													break;
											}
											else {
												if(j < jobBandWidthList.size() -1) {
													logger.info("continue");
													continue;
												}
												if(j == jobBandWidthList.size() - 1) {
													logger.info("the last one compareTo");
													newList.add(newBand);
												}
											}
										}
									}
									
									logger.info("newList: " + newList.size());
									if(!CollectionUtils.isEmpty(updateList)) {
										logger.info("updateList: " + updateList.size());
									}
									userJobBandWidthService.updateList(updateList);
									userJobBandWidthService.insertList(newList);
								}
								else {
									userJobBandWidthService.insertList(insertList);
								}
							}
						}
					}
				} catch (Exception e) {
					logger.error("get prometheus error");
					logger.error(e.getMessage(), e);
					continue;
				}
			}
		}
		return 1;
		
	}
	
	
	@Override
	public void updateBandWidthRes() {
			List<PaiUserResourceInfo> updateList = Lists.newArrayList();
			List<Map<String, BigDecimal>> list = userJobBandWidthService.selectUserBandWidth();
			if(!CollectionUtils.isEmpty(list)) {
				for(Map map : list) {
					PaiUserResourceInfo info = new PaiUserResourceInfo();
					String userId = (String) map.get("user_id");
					//KB
					BigDecimal tapeWidthResource = (BigDecimal) map.get("band_width");
					info.setUserId(userId);
					tapeWidthResource = tapeWidthResource.divide((new BigDecimal(1024 * 1024)), 4, BigDecimal.ROUND_HALF_UP);
					//GB
					info.setBandwidthUsedResource(tapeWidthResource);
					updateList.add(info);
				}
			}
			if(!CollectionUtils.isEmpty(updateList)) {
				updateList(updateList);
			}
	}
	
	
	private String authToken() {
		String token = "";
		try {
			MultiValueMap<String, String> map = new LinkedMultiValueMap<>();
			map.add("username", "paictrlcenter@leinao.ai");
			map.add("password", "ughYCd74");
			Integer t = 1 * 2 * 60 * 60;
			map.add("expiration", t.toString());
			logger.info("auth start...");
			String resBody = restTemplate.postForObject(loginUrl + "/token", map, String.class);
			logger.info("request auth return: " + resBody);
			JSONObject reponse = JSON.parseObject(resBody);
			token = reponse.getString("token");
			logger.info(token);
		} catch (Exception e) {
			logger.error("token auth error");
			logger.error(e.getMessage(),e);
		}
		return token;
	}

	@Transactional
	@Override
	public List<PaiUserResourceInfo> insertList(List<PaiUserResourceInfo> list) {
		if (logger.isInfoEnabled()) {
			logger.info(" insert lists : " + (list == null ? "null" : list.size()));
		}
		List<PaiUserResourceInfo> resultList = Lists.newArrayList();

		if (CollectionUtils.isEmpty(list)) {
			return new ArrayList<PaiUserResourceInfo>();
		}
		
		long currentTimeMillis = System.currentTimeMillis();
		
		try {
			for (PaiUserResourceInfo info : list) {
				info.setCreateTime(currentTimeMillis);
				info.setUpdateTime(currentTimeMillis);
				Long id = paiUserResourceInfoMapper.insert(info);
				if(id > 0) {
					resultList.add(info);
				}
			}
			
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
		
		if (logger.isInfoEnabled()) {
			logger.info(" insert lists  success : " + (resultList == null ? "null" : resultList.size()));
		}
		
		return resultList;
	}
	
	@Transactional
	@Override
	public List<PaiUserResourceInfo> updateList(List<PaiUserResourceInfo> list) {
		if (logger.isInfoEnabled()) {
			logger.info(" update lists : " + (list == null ? "null" : list.size()));
		}
		List<PaiUserResourceInfo> resultList = Lists.newArrayList();
		
		if (CollectionUtils.isEmpty(list)) {
			return new ArrayList<PaiUserResourceInfo>();
		}
		
		long currentTimeMillis = System.currentTimeMillis();
		
		try {
			for (PaiUserResourceInfo info : list) {
				info.setUpdateTime(currentTimeMillis);
				int rowNum = paiUserResourceInfoMapper.update(info);
				if(rowNum > 0) {
					resultList.add(info);
				}
			}
			
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
		
		if (logger.isInfoEnabled()) {
			logger.info(" update lists  success : " + (resultList == null ? "null" : resultList.size()));
		}
		
		return resultList;
	}
}